package window

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time

object WindowsBasic {
  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val stream1: DataStream[Click] = env.socketTextStream("localhost", 9999)
      .map(line => {
        val strs = line.split(",")
        Click(strs(0).toInt, strs(1), strs(2).toLong, strs(3).toInt)
      })

    //
    //    //牛刀小试  测试count window
    //    stream1.countWindowAll(5,2)
    //      .sum("age")
    //      .print()

    //    stream1.windowAll(GlobalWindows.create())
    //        .trigger(CountTrigger.of(2))
    //        .evictor(CountEvictor.of(5))
    //        .sum("age")
    //        .print()

    //    stream1.windowAll(GlobalWindows.create())
    //        .trigger(new MyCountTrigger1())
    //        .evictor(new MyEvictor1())
    //        .sum("age")
    //        .print()

    //    stream1.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(3)))
    //      .sum("age")
    //      .print

    stream1.windowAll(SlidingProcessingTimeWindows.of(Time.seconds(5), Time.seconds(2)))
      //.sum("age")
      .apply(new MyWindowFunction)
      .print


    env.execute("window test")

  }

}

case class Click(uid: Int, action: String, dt: Long, age: Int)


